import json
import logging
import random
import re
import string
import subprocess
import time
import uuid
from collections.abc import Generator, Mapping
from datetime import datetime
from hashlib import sha256
from typing import TYPE_CHECKING, Any, Optional, Union, cast
from zoneinfo import available_timezones

from flask import Response, stream_with_context
from flask_restful import fields

# from configs import dify_config # Removed Dify specific config import
import config # Use local config
from core.app.features.rate_limiting.rate_limit import RateLimitGenerator # This might need further refactoring or mocking if it has Dify dependencies
from core.file import helpers as file_helpers # This might need further refactoring or mocking if it has Dify dependencies
from extensions import redis_client # Use local extensions

if TYPE_CHECKING:
    from models.account import Account # This might need further refactoring or mocking if it has Dify dependencies


def run(script):
    return subprocess.getstatusoutput("source /root/.bashrc && " + script)


class AppIconUrlField(fields.Raw):
    def output(self, key, obj):
        if obj is None:
            return None

        # from models.model import App, IconType, Site # Dify specific model, may need to be refactored or mocked
        # For standalone, we might not have these complex App/Site models directly
        # This part needs careful review based on actual usage in the refactored service
        # For now, returning None to avoid errors if these models are not present
        return None


class AvatarUrlField(fields.Raw):
    def output(self, key, obj):
        if obj is None:
            return None

        # from models.account import Account # Dify specific model
        # Similar to AppIconUrlField, this needs review for standalone usage
        # if isinstance(obj, Account) and obj.avatar is not None:
        #     return file_helpers.get_signed_file_url(obj.avatar)
        return None


class TimestampField(fields.Raw):
    def format(self, value) -> int:
        return int(value.timestamp())


def email(email_str: str):
    pattern = r"^[\w\.!#$%&'*+\-/=?^_`{|}~]+@([\w-]+\.)+[\w-]{2,}$";
    if re.match(pattern, email_str) is not None:
        return email_str

    error = "{email} is not a valid email.".format(email=email_str)
    raise ValueError(error)


def uuid_value(value):
    if value == "":
        return str(value)

    try:
        uuid_obj = uuid.UUID(value)
        return str(uuid_obj)
    except ValueError:
        error = "{value} is not a valid uuid.".format(value=value)
        raise ValueError(error)


def alphanumeric(value: str):
    if re.match(r"^[a-zA-Z0-9_]+$", value):
        return value

    raise ValueError(f"{value} is not a valid alphanumeric value")


def timestamp_value(timestamp):
    try:
        int_timestamp = int(timestamp)
        if int_timestamp < 0:
            raise ValueError
        return int_timestamp
    except ValueError:
        error = "{timestamp} is not a valid timestamp.".format(timestamp=timestamp)
        raise ValueError(error)


class StrLen:
    def __init__(self, max_length, argument="argument"):
        self.max_length = max_length
        self.argument = argument

    def __call__(self, value):
        length = len(value)
        if length > self.max_length:
            error = "Invalid {arg}: {val}. {arg} cannot exceed length {length}".format(
                arg=self.argument, val=value, length=self.max_length
            )
            raise ValueError(error)

        return value


class FloatRange:
    def __init__(self, low, high, argument="argument"):
        self.low = low
        self.high = high
        self.argument = argument

    def __call__(self, value):
        value = _get_float(value)
        if value < self.low or value > self.high:
            error = "Invalid {arg}: {val}. {arg} must be within the range {lo} - {hi}".format(
                arg=self.argument, val=value, lo=self.low, hi=self.high
            )
            raise ValueError(error)

        return value


class DatetimeString:
    def __init__(self, format_str, argument="argument"):
        self.format = format_str
        self.argument = argument

    def __call__(self, value):
        try:
            datetime.strptime(value, self.format)
        except ValueError:
            error = "Invalid {arg}: {val}. {arg} must be conform to the format {format}".format(
                arg=self.argument, val=value, format=self.format
            )
            raise ValueError(error)

        return value


def _get_float(value):
    try:
        return float(value)
    except (TypeError, ValueError):
        raise ValueError("{} is not a valid float".format(value))


def timezone(timezone_string):
    if timezone_string and timezone_string in available_timezones():
        return timezone_string

    error = "{timezone_string} is not a valid timezone.".format(timezone_string=timezone_string)
    raise ValueError(error)


def generate_string(n):
    letters_digits = string.ascii_letters + string.digits
    result = ""
    for i in range(n):
        result += random.choice(letters_digits)

    return result


def extract_remote_ip(request) -> str:
    if request.headers.get("CF-Connecting-IP"):
        return cast(str, request.headers.get("Cf-Connecting-Ip"))
    elif request.headers.getlist("X-Forwarded-For"):
        return cast(str, request.headers.getlist("X-Forwarded-For")[0])
    else:
        return cast(str, request.remote_addr)


def generate_text_hash(text: str) -> str:
    hash_text = str(text) + "None"
    return sha256(hash_text.encode()).hexdigest()


def compact_generate_response(response: Union[Mapping, Generator, RateLimitGenerator]) -> Response:
    if isinstance(response, dict):
        return Response(response=json.dumps(response), status=200, mimetype="application/json")
    else:
        def generate() -> Generator:
            yield from response
        return Response(stream_with_context(generate()), status=200, mimetype="text/event-stream")


class TokenManager:
    @classmethod
    def generate_token(
        cls,
        token_type: str,
        account: Optional[Any] = None, # Changed from "Account" to Any for standalone
        email: Optional[str] = None,
        additional_data: Optional[dict] = None,
    ) -> str:
        if account is None and email is None:
            raise ValueError("Account or email must be provided")

        # account_id = account.id if account else None # Dify specific account model
        # account_email = account.email if account else email # Dify specific account model
        # For standalone, we might simplify this or use a different user identification mechanism
        account_id = getattr(account, 'id', None) if account else None
        account_email = getattr(account, 'email', email) if account else email


        if account_id:
            old_token = cls._get_current_token_for_account(account_id, token_type)
            if old_token:
                if isinstance(old_token, bytes):
                    old_token = old_token.decode("utf-8")
                cls.revoke_token(old_token, token_type)

        token = str(uuid.uuid4())
        token_data = {"account_id": account_id, "email": account_email, "token_type": token_type}
        if additional_data:
            token_data.update(additional_data)

        # Get expiry from local config, with a default
        expiry_minutes_attr = f"{token_type.upper()}_TOKEN_EXPIRY_MINUTES"
        expiry_minutes = getattr(config, expiry_minutes_attr, 60) # Default to 60 minutes if not in config
        
        if expiry_minutes is None: # Should not happen with default
            raise ValueError(f"Expiry minutes for {token_type} token is not set in config.py")
        token_key = cls._get_token_key(token, token_type)
        expiry_time = int(expiry_minutes * 60)
        redis_client.set(token_key, json.dumps(token_data), ex=expiry_time) # Use set with ex for expiry

        if account_id:
            cls._set_current_token_for_account(account_id, token, token_type, expiry_minutes)

        return token

    @classmethod
    def _get_token_key(cls, token: str, token_type: str) -> str:
        return f"{token_type}:token:{token}"

    @classmethod
    def revoke_token(cls, token: str, token_type: str):
        token_key = cls._get_token_key(token, token_type)
        redis_client.delete(token_key)

    @classmethod
    def get_token_data(cls, token: str, token_type: str) -> Optional[dict[str, Any]]:
        key = cls._get_token_key(token, token_type)
        token_data_json = redis_client.get(key)
        if token_data_json is None:
            logging.warning(f"{token_type} token {token} not found with key {key}")
            return None
        if isinstance(token_data_json, bytes):
            token_data_json = token_data_json.decode('utf-8')
        token_data: Optional[dict[str, Any]] = json.loads(token_data_json)
        return token_data

    @classmethod
    def _get_current_token_for_account(cls, account_id: str, token_type: str) -> Optional[str]:
        key = cls._get_account_token_key(account_id, token_type)
        current_token_bytes: Optional[bytes] = redis_client.get(key)
        if current_token_bytes:
            return current_token_bytes.decode('utf-8')
        return None

    @classmethod
    def _set_current_token_for_account(
        cls, account_id: str, token: str, token_type: str, expiry_minutes: Union[int, float]
    ):
        key = cls._get_account_token_key(account_id, token_type)
        expiry_time = int(expiry_minutes * 60)
        redis_client.set(key, token, ex=expiry_time) # Use set with ex for expiry

    @classmethod
    def _get_account_token_key(cls, account_id: str, token_type: str) -> str:
        return f"{token_type}:account:{account_id}"


class RateLimiter:
    def __init__(self, prefix: str, max_attempts: int, time_window: int):
        self.prefix = prefix
        self.max_attempts = max_attempts
        self.time_window = time_window

    def _get_key(self, email: str) -> str:
        return f"{self.prefix}:{email}"

    def is_rate_limited(self, email: str) -> bool:
        key = self._get_key(email)
        # MockRedisClient does not have zremrangebyscore or zcard, so we simplify this logic
        # This is a very basic rate limiting mock for standalone usage.
        # For a real implementation, a proper Redis client and logic would be needed.
        attempts_key = f"{key}:attempts"
        current_attempts_bytes = redis_client.get(attempts_key)
        current_attempts = int(current_attempts_bytes.decode('utf-8')) if current_attempts_bytes else 0
        return current_attempts >= self.max_attempts

    def increment_rate_limit(self, email: str):
        key = self._get_key(email)
        attempts_key = f"{key}:attempts"
        # MockRedisClient does not have zadd or expire, so we simplify this logic
        current_attempts_bytes = redis_client.get(attempts_key)
        current_attempts = int(current_attempts_bytes.decode('utf-8')) if current_attempts_bytes else 0
        
        if current_attempts == 0:
            redis_client.set(attempts_key, str(current_attempts + 1), ex=self.time_window)
        else:
            # For a real redis, you would use INCR and check TTL, or use sorted sets as in original.
            # This mock will just increment if key exists, relying on the initial EXPIRE.
            # This is not a perfect replacement for the original sorted set logic.
            redis_client.set(attempts_key, str(current_attempts + 1)) # No TTL update here for simplicity

